package edu.upenn.seas.peerreview.transport

import scala.actors._
import scala.actors.Actor._
import scala.actors.remote._
import scala.collection.mutable
import scala.collection.immutable.Queue
import scala.util.logging.Logged

import edu.upenn.seas.peerreview.{Id, Message}

/**
 * A remote (multi-JVM, centralized) transport layer's server component. Sources
 * are identified using (client id, local id) pairs so that a message can be
 * first routed to the right client and then dispatched using a local transport.
 */
class RemoteServer (port: Int) extends Transport.Server with Logged {
  
  import RemoteServer._
  import Transport._
  
  // TODO: factor out "set up remote actor" code
  // TODO: figure out why RemoteClientTest: testMultipleClassLoaders
  // occasionally receives None in the server's response.
  // TODO: make the mesgThread definition look pretty
  val mesgThread = actor {
    val cLoader = Thread.currentThread.getContextClassLoader
    RemoteActor.classLoader = cLoader
    RemoteActor.alive(port)
    RemoteActor.register('server, self)
    log("self uses port %s, cLoader %s".format(port, cLoader))
    
    var coordinatorClientId: ClientId = null
    val idToMessageQueue = mutable.Map[ClientId,mutable.Queue[SendMessage]]()
    var coordinatorWaitingQueue = new mutable.Queue[SendMessage]()

    def validateClient(clientId: ClientId, response: Any=>Unit): Boolean =
      if (!(idToMessageQueue isDefinedAt clientId)) {
        response(InvalidClientId(clientId))
        false
      } else {
        true
      }
    
    loop {
      react {
        case m @ RegisterClient() => {
          log("received " + m)
          val clientId = ClientId.newClientId()
          idToMessageQueue(clientId) = new mutable.Queue()
          reply(RegisteredClient(clientId))
          log("assigned %s to newly registered client" format clientId)
        }
        case m @ RegisterCoordinator(clientId: ClientId) => {
          log("received " + m)
          if (coordinatorClientId == null) {
            if (validateClient(clientId, reply)) {
              log("coordinator is unclaimed, giving claim to " + clientId)
              coordinatorClientId = clientId
              idToMessageQueue(coordinatorClientId) ++= coordinatorWaitingQueue
              coordinatorWaitingQueue = null // prevent accidental future use
            }         
          }
          reply(RegisteredCoordinator(coordinatorClientId == clientId))
          log("RegisterCoordinator result " + (coordinatorClientId == clientId))
        }
        case m @ SendMessage(from,to, mesg) => {
          log("received " + m)
          if (to != Id.coordinatorId) {
            val clientId = to.asInstanceOf[RemoteId].clientId
            if (validateClient(clientId, reply)) {
              log("enqueued message with client queue")
              idToMessageQueue(clientId) += m
            }     
          } else if (coordinatorClientId == null) {
            log("enqueued message with coordinator waiting queue");
            coordinatorWaitingQueue += m
          } else {
            log("enqueued message with queue of client that claimed coord")
            idToMessageQueue(coordinatorClientId) += m
          }
        }
        case m @ RequestMessages(clientId) => {
          if (validateClient(clientId, reply)) {
            val messages = List() ++ idToMessageQueue(clientId)
            idToMessageQueue(clientId).clear
            reply(RequestedMessages(messages))
          }     
        }
        case m => log("uh oh, received unrecognized message " + m)
      }
    }
  }

  def stop() = try {
    log("trying to stop the server thread")
    mesgThread.exit()
  } catch {
    case ex => log("exception when stopping the server thread: " + ex)
  }
}

object RemoteServer {

  /**
   * Identifies a non-coordinator source.
   */
  case class RemoteId (clientId: ClientId, localId: Id) extends Id {
    assert(localId != Id.coordinatorId, "Error: shouldn't wrap coord id.")
    override def toString = "%s.%s" format (clientId, localId)
  }

  /**
   * Identifies a client component in the remote transport layer.
   */
  case class ClientId private(id: Double) {
    override def toString = "Client%d" format id.floor.toInt
  }

  object ClientId {
    var nextClientId = 0
    
    /**
     * Not thread safe. Uses randomness to distinguish between clients
     * registered with different server instances.
     */
    def newClientId(): ClientId = {
      nextClientId += 1
      new ClientId(nextClientId + Math.random)
    }
  }

  /**
   * Sent to the server to request a client id.
   * @see RegisteredClient
   */
  case class RegisterClient()
  
  /**
   * Sent to a client to assign a client id.
   * @see RegisterClient
   */
  case class RegisteredClient(clientId: ClientId)

  /**
   * Sent to the server to request permission to register the coordinator.
   * @see RegisteredCoordinator
   */
  case class RegisterCoordinator(clientId: ClientId)

  /**
   * Sent to a client to grant or deny permission to register the coordinator.
   * @param permissionGranted true iff the server grants permission
   * @see RegisterCoordinator
   */
  case class RegisteredCoordinator(permissionGranted: Boolean)

  /**
   * Sent to the server to enqueue a message.
   * @see RequestMessages
   */
  case class SendMessage(from: Id, to: Id, mesg: Message) {
    override def toString =
      "SendMessage(%s from %s to %s)".format(mesg, from, to)
  }

  /**
   * Sent to the server to dequeue messages sent to sources at the given client.
   * @see RequestedMessages
   */
  case class RequestMessages(forDestination: ClientId)

  /**
   * Sent to a client to dequeue messages sent to sources at a given client.
   * @param mesgs the list of messages, to be dispatched <b>in order</b>
   * @see RequestMessages
   */
  case class RequestedMessages(mesgs: List[SendMessage])

  /**
   * Sent to a client to indicate that the server received an invalid client id.
   * @see ClientId
   */
  case class InvalidClientId(clientId : ClientId)
}
